/**
 * Copyright 2009 the original author or authors.
 *
 * <p>Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file
 * except in compliance with the License. You may obtain a copy of the License at
 *
 * <p>http://www.apache.org/licenses/LICENSE-2.0
 *
 * <p>Unless required by applicable law or agreed to in writing, software distributed under the
 * License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
 * express or implied. See the License for the specific language governing permissions and
 * limitations under the License.
 */
package net.javacrumbs.shedlock.provider.cassandra;

import static com.datastax.oss.driver.api.core.CqlIdentifier.fromCql;
import static net.javacrumbs.shedlock.core.ClockProvider.now;
import static net.javacrumbs.shedlock.provider.cassandra.CassandraLockProvider.DEFAULT_TABLE;
import static net.javacrumbs.shedlock.test.support.DockerCleaner.removeImageInCi;
import static org.assertj.core.api.Assertions.assertThat;

import com.datastax.oss.driver.api.core.CqlIdentifier;
import com.datastax.oss.driver.api.core.CqlSession;
import com.datastax.oss.driver.api.querybuilder.QueryBuilder;
import java.net.InetSocketAddress;
import net.javacrumbs.shedlock.provider.cassandra.CassandraLockProvider.Configuration;
import net.javacrumbs.shedlock.support.StorageBasedLockProvider;
import net.javacrumbs.shedlock.test.support.AbstractStorageBasedLockProviderIntegrationTest;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeAll;
import org.testcontainers.cassandra.CassandraContainer;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;

/**
 * Integration test uses local instance of Cassandra running on localhost at
 * port 9042 using keyspace shedlock and table lock
 *
 * @see net.javacrumbs.shedlock.provider.cassandra.CassandraLockProvider
 */
@Testcontainers
public class CassandraLockProviderIntegrationTest extends AbstractStorageBasedLockProviderIntegrationTest {
    private static final CqlIdentifier KEYSPACE = fromCql("shedlock");
    private static CqlSession session;
    private static final String DOCKER_IMAGE_NAME = "cassandra:4";

    @Container
    static final CassandraContainer cassandra = new CassandraContainer(DOCKER_IMAGE_NAME)
            .withInitScript("shedlock.cql")
            .withEnv("CASSANDRA_DC", "local")
            .withEnv("CASSANDRA_ENDPOINT_SNITCH", "GossipingPropertyFileSnitch");

    @BeforeAll
    public static void startCassandra() {
        String containerIpAddress = cassandra.getHost();
        int containerPort = cassandra.getMappedPort(9042);
        InetSocketAddress containerEndPoint = new InetSocketAddress(containerIpAddress, containerPort);

        session = CqlSession.builder()
                .addContactPoint(containerEndPoint)
                .withLocalDatacenter("local")
                .build();
    }

    @AfterAll
    public static void removeImage() {
        removeImageInCi(DOCKER_IMAGE_NAME);
    }

    @AfterEach
    public void after() {
        session.execute(QueryBuilder.truncate(KEYSPACE, fromCql(DEFAULT_TABLE)).build());
    }

    @Override
    protected StorageBasedLockProvider getLockProvider() {
        return new CassandraLockProvider(Configuration.builder()
                .withCqlSession(session)
                .withKeyspace(KEYSPACE)
                .build());
    }

    @Override
    protected void assertUnlocked(String lockName) {
        Lock lock = findLock(lockName);
        assertThat(lock.lockUntil()).isBefore(now());
        assertThat(lock.lockedAt()).isBefore(now());
        assertThat(lock.lockedBy()).isNotEmpty();
    }

    @Override
    protected void assertLocked(String lockName) {
        Lock lock = findLock(lockName);
        assertThat(lock.lockUntil()).isAfter(now());
        assertThat(lock.lockedAt()).isBefore(now());
        assertThat(lock.lockedBy()).isNotEmpty();
    }

    private Lock findLock(String lockName) {
        CassandraStorageAccessor cassandraStorageAccessor = new CassandraStorageAccessor(Configuration.builder()
                .withCqlSession(session)
                .withKeyspace(KEYSPACE)
                .withTableName(DEFAULT_TABLE)
                .build());
        return cassandraStorageAccessor.find(lockName).get();
    }
}
